package Batch::Cluster;
use Moose;
use GRID::Machine;
use Batch::Cluster::Result;
use IO::Select;

our $VERSION = '0.01';

# Guarda los objetos GRID::Machine.
has 'machines' => (
    traits => ['Hash'],
    is => 'rw',
    isa => 'HashRef[Any]',
    default => sub { {} },
    handles => {
      set_machine => 'set',
      get_machine => 'get',
    },
);

# Guarda los nombres de las máquinas manteniendo el orden dado.
has 'hosts' => (
    traits => ['Array'],
    is => 'rw',
    isa => 'ArrayRef[Str]',
    default => sub { [] },
    handles => {
      list_machines => 'elements',
    },
);


around BUILDARGS => sub {
  my $orig = shift;
  my $class = shift;

  my (@hosts, %machines);
  
  for (@_) {
    eval {
      $machines{$_} = GRID::Machine->new( host => $_ );
      push @hosts, $_;
    };
    if ($@) {
      warn "Aviso: No se pudo conectar al host '$_': $@";
    }
  }
  return undef unless @hosts;

  return $class->$orig(machines => \%machines, hosts => \@hosts);
};


sub run {
  my $self = shift;
  my @commands = map { "$_ | " } @_;

  my %id;
  my $np = @commands;
  my $lp = $np - 1;
  my $cont = 0;
  my $sel = IO::Select->new();

  # Manda la primera tanda de trabajos a todos los nodos.
  for ($self->list_machines()) {
    my $m = $self->get_machine($_);
    my $handler = $m->open(shift @commands);
    $id{scalar $handler} = [$m, $cont++];
    $sel->add($handler);
    last if ($cont > $lp);
  }

  my $result = Batch::Cluster::Result->new();
  my @ready;
  my $aux = 0;

  # Espera a que vayan dando salida para capturarla.
  do {
    push @ready, $sel->can_read unless @ready;
    my $handler = shift @ready;
    my ($m, $pos) = @{$id{$handler}};
    
    my ($aux, $r);

    # Obtiene la salida del proceso remoto.
    $r .= $aux while (sysread($handler, $aux, 4));
    chomp $r if $r;

    $sel->remove($handler) if eof($handler);
    close($handler);
    
    $result->add_ordered_result(GRID::Machine::Result->new( results => [ $r ] ), $pos);

    # Después de que se haya procesado al salida del nodo se le asinga un nuevo trabajo.
    if (@commands) {
      my $h = $m->open(shift @commands);
      $id{scalar $h} = [$m, $cont++];
      $sel->add($h);
    }

  } until (++$aux == $np);

  # Si estamos en contexto de lista se devuelven los resultados en una lista.
  return map { $_->result() } $result->all_results() if (wantarray);

  # En contexto escalar se devuelve el objeto Batch::Cluster::Result.
  return $result;
}

sub copyandmake {
  my ($self, %args) = @_;
  $self->get_machine($_)->copyandmake(%args) for $self->list_machines();
}

sub chdir {
  my ($self, $dir) = @_;
  my $result = Batch::Cluster::Result->new();
  $result->add_result($self->get_machine($_)->chdir($dir)) for $self->list_machines();
  return $result;
}

__PACKAGE__->meta->make_immutable;

no Moose;

1;

__END__

=head1 NAME

Batch::Cluster - Módulo para la ejecución de programas en un cluster remoto a través de SSH.

=head1 SYNOPSIS

  use Batch::Cluster;
  use List::Util qw{sum};

  my $np = 5;
  my $N = 50000;
  my $clean = 1;

  my $c = Batch::Cluster->new(@hostnames) || die "Error Intializing Cluster";

  # Transference of files to remote hosts
  $c->copyandmake(
    dir => 'pi',
    makeargs => 'pi',
    files => [ qw{etc/pi/pi.c etc/pi/Makefile} ],
    cleanfiles => $clean,
    cleandirs => $clean, # remove the whole directory at the end
    keepdir => 1,
  );

  $c->chdir("pi/")  || die "Can't change to pi/\n";

  my @commands = map {  "./pi $_ $N $np" } 0..$np-1;

  my $pi = sum $c->run(@commands);

=head1 DESCRIPTION

El módulo Batch::Cluster hace uso del módulo GRID::Machines para crear un
cluster de máquinas remotas a través de SSH con el fin de realizar ejecuciones
de programas de forma sencila. El cluster es mostrado de forma unitaria, de
forma que se puedan realizar acciones sobre todo el cluster como si de una sola
máquina se tratara.

=head1 AUTHOR

Juan Pablo Chinea Martin, E<lt>alu2592@etsii.ull.esE<gt>

=head1 COPYRIGHT AND LICENSE

Copyright (C) 2011 by Juan Pablo Chinea Martin

This library is free software; you can redistribute it and/or modify
it under the same terms as Perl itself, either Perl version 5.10.0 or,
at your option, any later version of Perl 5 you may have available.


=cut
